package com.atguigu.bigdata.gmall.realtime.app.dwd;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_02_DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdTradeOrderDetail().init(
                2002,
                2,
                "Dwd_02_DwdTradeOrderDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db 的数据
        readOdsDb(tEnv, "Dwd_02_DwdTradeOrderDetail");
        // 2.过滤出详情表
        Table orderDetail = tEnv.sqlQuery(" select" +
                " `data`['id'] id, " +
                " `data`['course_id'] course_id, " +
                " `data`['course_name'] course_name, " +
                " `data`['order_id'] order_id, " +
                " ts, " +
                " pt " +
                " from ods_db" +
                " where `database`='gmall2022'" +
                " and `table`='order_detail'" +
                " and `type`='insert'");
        tEnv.createTemporaryView("order_detail", orderDetail);
        // 3.过滤出订单表
        Table orderInfo = tEnv.sqlQuery(" select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['province_id'] province_id, " +
                " `data`['origin_amount'] origin_amount, " +
                " `data`['final_amount'] final_amount, " +
                " `data`['order_status'] order_status " +
                " from ods_db " +
                " where `database`='gmall2022' " +
                " and `table`='order_info' " +
                " and `type`='insert'");
        tEnv.createTemporaryView("order_info", orderInfo);
//        orderInfo.execute().print();


/*        // 4.过滤出课程信息表
        Table courseInfo = tEnv.sqlQuery(" select" +
                "  `data`['id'] id," +
                "  `data`['subject_id'] subject_id" +
                " from ods_db" +
                " where `database`='gmall2022'" +
                " and `table`='course_info'" +
                " and `type`='insert'");
        tEnv.createTemporaryView("course_info", courseInfo);
        courseInfo.execute().print();
        // 5.过滤出科目表
        Table baseSubjectInfo = tEnv.sqlQuery(" select" +
                "  `data`['id'] id," +
                "  `data`['category_id'] category_id " +
                "  from ods_db" +
                "  where `database`='gmall2022'" +
                "  and `table`='base_subject_info'" +
                "  and `type`='insert' ");
        tEnv.createTemporaryView("base_subject_info", baseSubjectInfo);
        baseSubjectInfo.execute().print();

        // 6.过滤出分类表
        Table baseCategoryInfo = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['category_name'] category_name " +
                "from ods_db " +
                "where `database`='gmall2022' " +
                "and `table`='base_category_info' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("base_category_info", baseCategoryInfo);
        baseCategoryInfo.execute().print();*/
        // 6.五张表的 join
        Table result = tEnv.sqlQuery("  select" +
                "  od.id," +
                "  od.course_id," +
                "  od.course_name," +
                "  od.order_id," +
                "  oi.user_id," +
                "  oi.province_id," +
                "  oi.origin_amount," +
                "  oi.final_amount," +
                "  oi.order_status," +
                "  od.ts," +
                "  current_row_timestamp() row_op_ts " +
                "  from order_detail od " +
                "  join order_info oi" +
                "  on od.order_id=oi.id");
//        result.execute().print();

        // 7.写入到 kafka
        tEnv.executeSql(" create table dwd_trade_order_detail(" +
                "  id string," +
                "  course_id string," +
                "  course_name string," +
                "  order_id string," +
                "  user_id string," +
                "  province_id string," +
                "  origin_amount string," +
                "  final_amount string," +
                "  order_status string," +
                "  ts bigint," +
                "  row_op_ts timestamp_ltz(3)," +
                "  primary key(id) not enforced" +
                "  )"+ SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        result.executeInsert("dwd_trade_order_detail");


    }
}
